package com.xzx.flink.streamapi.watermark;

import com.xzx.flink.bean.ClickEvent;
import com.xzx.flink.streamapi.source.ClickSource;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 周期性的水位线
 *
 * @version 1.0
 * @auther xinzhixuan
 * @date 2022/4/30 13:58
 */
public class Watermark_Periodic {
    private static final Logger log = LoggerFactory.getLogger(Watermark_Periodic.class);

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.getConfig().setAutoWatermarkInterval(5000);// 5秒触发一次水位线，默认是200毫秒

        env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(new CustomPeriodicWatermarkStrategy())
                .print();

        env.execute(Watermark_Periodic.class.getSimpleName());
    }

    public static class CustomPeriodicWatermarkStrategy implements WatermarkStrategy<ClickEvent> {

        // 时间戳分配器
        @Override
        public TimestampAssigner<ClickEvent> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
            return new SerializableTimestampAssigner<ClickEvent>() {
                @Override
                public long extractTimestamp(ClickEvent element, long recordTimestamp) {
                    return element.getTimestamp();
                }
            };
        }

        // 水位线生成器
        @Override
        public WatermarkGenerator<ClickEvent> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
            return new PeriodicWatermarkGenerator();
        }
    }

    public static class PeriodicWatermarkGenerator implements WatermarkGenerator<ClickEvent> {
        private static final Long DELAY = 2000L;// 延迟2秒
        private long currentMaxEventTimestamp = Long.MIN_VALUE;
        //每来一条就会触发一次这个方法
        @Override
        public void onEvent(ClickEvent event, long eventTimestamp, WatermarkOutput output) {
            log.info("=============onEvent====");
            currentMaxEventTimestamp = Math.max(currentMaxEventTimestamp, eventTimestamp - DELAY - 1);
        }

        // 这个方法会被系统调用，默认是200毫秒调用一次
        @Override
        public void onPeriodicEmit(WatermarkOutput output) {
            log.info("=============onPeriodicEmit====");
            output.emitWatermark(new Watermark(currentMaxEventTimestamp));
        }
    }
}
